package com.baishancloud.log.common.sink

import com.baishancloud.log.common.connector.es.ESSinkConnector7
import com.baishancloud.log.common.connector.kafka.KafkaSinkConnector
import com.baishancloud.log.common.connector.starRocks.StarRocksSinkConnector
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer
import org.apache.kafka.clients.producer.ProducerRecord

/**
 *
 * @author ziqiang.wang
 * @date 2021/10/21 15:38
 */
object SinkUtil extends Serializable {


  /**
   * 动态索引<br>
   * 返回可以将json字符串写入动态索引的es的sink连接器<br>
   * 接收二元组(indexSuffix,data)
   *
   * @param parameterTool 主类入参对象
   */
  def esDynamicIndex(parameterTool: ParameterTool, number: String = ""): ElasticsearchSink[(String, String)] = {
    ESSinkConnector7.builder(parameterTool, number).dynamicIndex()
  }

  /**
   * 动态索引<br>
   * 返回可以将json字符串写入动态索引的es的sink连接器，同时支持索引id，覆盖形式写入<br>
   * 接收三元组(id,indexSuffix,data)
   *
   * @param parameterTool 主类入参对象
   */
  def esDynamicIndexHaveId(parameterTool: ParameterTool, number: String = ""): ElasticsearchSink[(String, String, String)] = {
    ESSinkConnector7.builder(parameterTool, number).dynamicIndexHaveId()
  }

  /**
   * 返回可以将json字符串写入starRocks的sink连接器
   *
   * @param parameterTool 主类入参对象
   * @param number        第几个sink
   */
  def starRocksJsonString(parameterTool: ParameterTool, number: String = ""): SinkFunction[String] = {
    StarRocksSinkConnector.builder(parameterTool, number).build().sinkJsonString()
  }

  /**
   * 返回只写入value的kafka sink
   *
   * @param parameterTool 主类入参对象
   * @param number        第几个sink
   */
  def kafkaOnlyValue(parameterTool: ParameterTool, number: String = ""): FlinkKafkaProducer[String] = {
    KafkaSinkConnector.builder(parameterTool, number).build().sinkOnlyValue()
  }

  /**
   * 返回写入key、value的kafka sink
   *
   * @param parameterTool 主类入参对象
   * @param number        第几个sink
   */
  def kafkaKeyValue(parameterTool: ParameterTool, number: String = ""): FlinkKafkaProducer[(String, String)] = {
    KafkaSinkConnector.builder(parameterTool, number).build().sinkKeyValue()
  }

  /**
   * 返回只写入value的kafka sink
   *
   * @param parameterTool 主类入参对象
   * @param number        第几个sink
   */
  def kafkaProducerRecord(parameterTool: ParameterTool, number: String = ""): FlinkKafkaProducer[ProducerRecord[Array[Byte], Array[Byte]]] = {
    KafkaSinkConnector.builder(parameterTool, number).build().sinkProducerRecord()
  }

  /**
   * 返回写入 mallard 的 sink
   *
   * @param outThreshold 写入mallard阈值，默认为int最大值
   * @param outInterval  写入mallard时间间隔，默认5秒，单位：秒
   */
  def mallard(outThreshold: Int = Int.MaxValue, outInterval: Long = 5): MallardSink = {
    new MallardSink(outThreshold, outInterval)
  }


}
